package com.gitee.xhs.keyedState.reducingState;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import static com.gitee.xhs.StateConstant.FLINK_CHECK_POINT_PATH;

/**
 * @author BigHead
 * @date 2022/6/30 13:52
 * @description ReducingStateConstant
 */
public class ReducingStateConstant {

    public static final String CUSTOM_REDUCING_STATE_DIR = "reducingState/";

    public static final String REDUCING_STATE_PATH = FLINK_CHECK_POINT_PATH + CUSTOM_REDUCING_STATE_DIR;
    public static final String REDUCING_STATE_UID = "reducing-state-uid";
    public static final String REDUCING_STATE_NAME = "reducing-state";
    public static final ReducingStateDescriptor<Double> REDUCING_STATE_DESCRIPTOR = new ReducingStateDescriptor<Double>(
            REDUCING_STATE_NAME,
            new CustomReducingFunction(),
            TypeInformation.of(Double.class)
    );

    private static class CustomReducingFunction implements ReduceFunction<Double> {

        @Override
        public Double reduce(Double value1, Double value2) throws Exception {
            return value1 + value2;
        }
    }

}
